package streamAPI.checkpoint;

import com.google.common.collect.Lists;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import util.model.WordCount;

/**
 * @author : yue.cao
 * @since : 2021-06-25
 */
public class TT {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Integer> source = env.fromCollection(Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9,3,1,6));
		source.keyBy(x -> x).process(new KeyedProcessFunction<Integer, Integer, String>() {

			private MapState<String, WordCount> cache;

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);

				MapStateDescriptor<String, WordCount> idVisitStateDescriptor = new MapStateDescriptor<>(
						"ttt-descriptor",
						String.class, WordCount.class);
				StateTtlConfig stateTTLConfig = StateTtlConfig
						.newBuilder(org.apache.flink.api.common.time.Time.days(1))
						.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
						.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
						.cleanupInRocksdbCompactFilter(1000000)
						.build();
				idVisitStateDescriptor.enableTimeToLive(stateTTLConfig);
				cache = getRuntimeContext().getMapState(idVisitStateDescriptor);
			}

			@Override
			public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {

				String key = "CY-" + value;
				WordCount wordCount = cache.get(key);
				if (wordCount == null) {
					wordCount = new WordCount(key, 1L);
					cache.put(key, wordCount);
				} else {
					wordCount.count = wordCount.count + 1L;
					cache.put(key, wordCount);
				}
				out.collect(wordCount.toString());
			}
		}).print();

		env.execute();
	}
}
